package org.qee.activemq;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import java.util.function.Consumer;

public class ActiveMqService {
    private String topicName;

    private Consumer<Message> consumer;

    private Destination destination;

    public ActiveMqService(String topicName, Consumer<Message> consumer) {
        this.topicName = topicName;
        this.consumer = consumer;
        if (topicName.startsWith("topic")) {
            destination = new ActiveMQTopic(topicName);
        } else {
            destination = new ActiveMQQueue(topicName);
        }
    }

    public Session createSession() throws JMSException {
        // 创建连接和会话
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        ActiveMQConnection conn = (ActiveMQConnection) factory.createConnection();
        conn.start();
        return conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }


    public void startConsumer() throws JMSException {
        Session session = createSession();
        // 创建消费者
        MessageConsumer messageConsumer = session.createConsumer(destination);
        messageConsumer.setMessageListener(consumer::accept);
    }

    public Producer producer() throws JMSException {
        Session session = createSession();
        return new Producer(session.createProducer(destination), session);
    }


    public static void main(String[] args) throws JMSException, InterruptedException {
        ActiveMqService activeMqService = new ActiveMqService("topic.abc", System.out::println);
        activeMqService.startConsumer();
        Producer producer = activeMqService.producer();
        int index = 0;
        while (index++ < 100) {
            producer.send(index + " message.");
        }


        ActiveMqService activeMqService2 = new ActiveMqService("queue.abc", System.out::println);
        activeMqService2.startConsumer();
        Producer producer2 = activeMqService2.producer();
        int index2 = 0;
        while (index2++ < 100) {
            producer2.send(index2 + " message.");
        }

        Thread.sleep(200000L);
    }
}
